package test;

import kafka.api.OffsetRequest;
import kafka.api.PartitionOffsetRequestInfo;
import kafka.common.ErrorMapping;
import kafka.common.TopicAndPartition;
import kafka.javaapi.*;
import kafka.javaapi.consumer.SimpleConsumer;

import java.util.*;

/**
 * Created by urfresh.mark on 2017/2/27.
 */
public class SimpleConsumerUtil {
    
    /**
     * 确定从(topic, partition)的什么地方开始读取消息，即确定offset
     * Kafka提供了2个常量：
     * `kafka.api.OffsetRequest.EarliestTime()`找到log data的开始时间
     * `kafka.api.OffsetRequest.LatestTime()`是最新的时间
     * <p>
     * 最旧的offset并不一定是0，因为随着时间的推移，部分数据将被从log中移除
     * <p>
     * <p>
     * 这里传入的参数`consumer`的host必须是(topic, partition)的leader，
     * 否则会报错，错误代码为6
     * <p>
     * 各种失败代码的含义可以查询这里：kafka.common.ErrorMapping
     */
    public long getLastOffset(String leaderHost, int port, String clientId,
                              String topic, int partition, long time) {
        /** 用leader host  创建一个SimpleConsumer */
        SimpleConsumer consumer = new SimpleConsumer(leaderHost, port, 100 * 1000, 64 * 1024, clientId);
        TopicAndPartition topicAndPartition = new TopicAndPartition(topic, partition);
        Map<TopicAndPartition, PartitionOffsetRequestInfo> reqInfo = new HashMap<>();
        reqInfo.put(topicAndPartition, new PartitionOffsetRequestInfo(time, 1));
        kafka.javaapi.OffsetRequest req =
                new kafka.javaapi.OffsetRequest(reqInfo, kafka.api.OffsetRequest.CurrentVersion(), clientId);
        OffsetResponse resp = consumer.getOffsetsBefore(req);
        consumer.close();
        /** 处理失败 */
        if (resp.hasError()) {
            short errorCode = resp.errorCode(topic, partition);
            System.err.println("为 (Topic=“" + topic + "”, Partition=“" + partition +
                    "”, Time=“" + time + "”) 查询 offset 失败，失败代码为：" + errorCode +
                    ",  失败原因为: " + ErrorMapping.exceptionFor(errorCode));
            return -1;
        }
        /** 为什么这里返回的是数组？数组中会有几个数据？
         *  经过实际的测试，里面只有1个数据 */
        long[] offsets = resp.offsets(topic, partition);
        return offsets[0];
    }
    
    
    /***
     * 测试`ConsumerB#getLastOffset`方法的使用
     * 找出(Topic=“topic-C”, Partition=1)的最老、最新、当前的offset
     * */
    public static void TestGetLastOffset() {
        List<String> brokers = new ArrayList<>();
//        brokers.add("ecs1");
//        brokers.add("ecs4");
        brokers.add("120.26.223.176");
        int port = 9092;
        String topic = "urfresh";
        int partition = 0;
        /**
         * 首先查询出(Topic=“topic-C”, Partition=“1”)的leader broker's hostname
         * */
        SimpleConsumerUtil util = new SimpleConsumerUtil();
        PartitionMetadata parMeta = util.findLeader(brokers, port, topic, partition);
        String leaderHost = parMeta.leader().host();
        /** 最老的offset（依然在log中的msg）*/
        long earliestOffset = util.getLastOffset(leaderHost, port, "",
                topic, partition, OffsetRequest.EarliestTime());
        /** 最新的offset */
        long latestOffset = util.getLastOffset(leaderHost, port, "",
                topic, partition, OffsetRequest.LatestTime());
        /** 当前的offset */
        long currentOffset = util.getLastOffset(leaderHost, port, "",
                topic, partition, System.currentTimeMillis());
        System.out.println("(Topic=“" + topic + "”, Partition=“" + partition +
                "”) 的leader host是" + leaderHost);
        System.out.println("(Topic=“" + topic + "”, Partition=“" + partition +
                "”) 中的offsets: " +
                "\n\tEarliest Offset: " + earliestOffset +
                "\n\tLatest   Offset: " + latestOffset +
                "\n\tCurrent  Offset: " + currentOffset);
    }
    
    /**
     * 为给定的(topic, partition)找到它的leader broker
     * seedBrokers并不一定要是全部的brokers，只要包含一个live broker，
     * 通过这个live broker去查询Leader broker的元数据即可
     */
    public PartitionMetadata
    findLeader(List<String> seedBrokers, int port, String topic, int partition) {
        PartitionMetadata retMeta = null;
        /** 循环地查询 */
        loop:
        for (String broker : seedBrokers) { /** 按照每一个 broker 循环*/
            SimpleConsumer consumer = null;
            try {
                /** 连接到不同的broker时，需要对该broker创建新的`SimpleConsumer`实例 */
                consumer = new SimpleConsumer(broker, port, 100 * 1000, 64 * 1024,
                        "leaderLookupClient");
                List<String> topics = Collections.singletonList(topic);
                TopicMetadataRequest req = new TopicMetadataRequest(topics);
                /** Fetch metadata for a sequence of topics, which
                 *  returns metadata for each  topic in the request */
                TopicMetadataResponse resp = consumer.send(req);
                /** 在某个broker上按照每一个topic循环，并检查一个topic中的每一个partition */
                for (TopicMetadata topicMeta : resp.topicsMetadata()) {
                    for (PartitionMetadata parMeta : topicMeta.partitionsMetadata()) {
                        if (parMeta.partitionId() == partition) {
                            retMeta = parMeta; /** 根据 partition id 进行匹配，找到了就退出 */
                            break loop;
                        }
                    }
                }
            } catch (Exception ex) {
                System.err.println("Error: 在向Broker “" + broker +
                        "” 询问 (Topic=“" + topic + "”, Partition=“" + partition
                        + "”) 的Leader Broker 时发生异常，原因为: \n" + ex.toString());
            } finally {
                if (null != consumer)
                    consumer.close(); /** 关闭连接到该broker的`SimpleConsumer`*/
            }
        } // end => for (String broker : seedBrokers)
        return retMeta;
    }
    
    public static void TestFindLeader() {
        SimpleConsumerUtil consumer = new SimpleConsumerUtil();
        List<String> seedBrokers = new ArrayList<>();
        seedBrokers.add("120.26.223.176");
//        seedBrokers.add("ecs3.njzd.com");
        PartitionMetadata parMeta = consumer.findLeader(seedBrokers, 9092, "urfresh", 0);
        System.out.println("Partition metadata for (Topic=“urfresh”, Partition=“0”) is : \n"
                + parMeta + "\n");
    }
    
    public static void main(String[] args) {
//        SimpleConsumerUtil.TestFindLeader();
    
        SimpleConsumerUtil.TestGetLastOffset();
    }
}
